package cn.lagou.sparkcore.work

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Work3 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    // 业务逻辑
    val clickLog = sc.textFile("./src/main/data/click.log")
    val impLog = sc.textFile("./src/main/data/imp.log")

    // 读取clicklog
    val clkRDD = clickLog.map { line =>
      val arr = line.split("\\s+")
      //截取获得adid
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, (1, 0))
    }
    // 读取implog
    val impRDD = impLog.map { line =>
      val arr = line.split("\\s+")
      //截取获得adid
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, (0, 1))
    }
    // 通过join将两个结果合并
    val RDD: RDD[(String, (Int, Int))] = clkRDD.union(impRDD)
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
    // 将结果写入hdfs
    RDD.saveAsTextFile("hdfs://linux121:9000/data/")
    // 关闭SparkContext
    sc.stop()

    /**
     * 3.2我的操作只有一个shuffle：
     *  因为读取日志时，map操作没有shuffle，而reduce操作有，而上面需要读取两个日志并且join，
     *  所以将reduce操作放到join之后可以减少shuffle次数
     */
  }

}
